package com.teemor.im.discovery.distributed;


import com.teemor.im.common.constants.ServerConstants;
import com.teemor.im.common.util.JsonUtil;
import com.teemor.im.common.util.ObjectUtil;
import com.teemor.im.discovery.node.LocalCacheNodeTable;
import com.teemor.im.discovery.node.ServerNode;
import com.teemor.im.discovery.zk.ZKclient;
import com.teemor.im.remoting.instance.NettyRemotingClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.data.Stat;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * create by 尼恩 @ 疯狂创客圈
 **/
@Slf4j
public class PeerManager {
    //Zk客户端
    private CuratorFramework client = null;

    private String pathRegistered = null;
    private ServerNode node = null;


    private static PeerManager singleInstance = null;
    private static final String path = ServerConstants.MANAGE_PATH;




    public static PeerManager getInst() {
        if (null == singleInstance) {
            singleInstance = new PeerManager();
            singleInstance.client = ZKclient.instance.getClient();
        }
        return singleInstance;
    }

    private PeerManager() {

    }


    /**
     * 初始化节点管理
     */
    public void init() {
        try {
            //获取已有节点
           List<String> nodePaths =  client.getChildren().forPath(path);
           nodePaths.forEach(nodePath -> {
               try {
                   nodePath = path + "/" +nodePath;
                   Stat stat = client.checkExists().forPath(nodePath);
                   if (stat!=null) {
                       byte[] nodeData = client.getData().forPath(nodePath);
                       processNodeAdded(nodeData, nodePath);
                   }
               } catch (Exception e) {
                   e.printStackTrace();
               }
           });


            //订阅节点的增加和删除事件

            PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
            PathChildrenCacheListener childrenCacheListener = (client, event) -> {
                log.info("开始监听其他的ImWorker子节点:-----");
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        log.info("CHILD_ADDED : " + data.getPath() + "  数据:" + data.getData());
                        processNodeAdded(data.getData(), data.getPath());
                        break;
                    case CHILD_REMOVED:
                        log.info("CHILD_REMOVED : " + data.getPath() + "  数据:" + data.getData());
                        processNodeRemoved(data);
                        break;
                    case CHILD_UPDATED:
                        log.info("CHILD_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
                        break;
                    default:
                        log.debug("[PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
                        break;
                }

            };

            childrenCache.getListenable().addListener(childrenCacheListener);
            System.out.println("Register zk watcher successfully!");
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void processNodeRemoved(ChildData data) {

        byte[] payload = data.getData();
        String info = new String(payload, StandardCharsets.UTF_8);
        ServerNode n = ObjectUtil.JsonBytes2Object(payload, ServerNode.class);

        String id = ImWorker.getInst().getIdByPath(data.getPath());
        n.setId(id);
        log.info("[TreeCache]节点删除, path={}, data={}",
                data.getPath(), JsonUtil.pojoToJson(n));
        LocalCacheNodeTable.removeNode(node);
    }

    private void processNodeAdded(byte[] payload , String path) {
        ServerNode n = ObjectUtil.JsonBytes2Object(payload, ServerNode.class);
        String id = ImWorker.getInst().getIdByPath(path);
        n.setId(id);
        log.info("[TreeCache]节点更新端口, path={}, data={}",
                path, JsonUtil.pojoToJson(n));

        if(n.equals(getLocalNode()))
        {
            log.info("[TreeCache]本地节点, path={}, data={}",
                    path, JsonUtil.pojoToJson(n));
            return;
        }
        LocalCacheNodeTable.addRemoteNode(n);
        NettyRemotingClient.instance().connect(n);

    }

    public ServerNode getLocalNode() {
        return ImWorker.getInst().getLocalNodeInfo();
    }

    public void remove(ServerNode remoteNode) {
       LocalCacheNodeTable.removeNode(remoteNode);
        log.info("[TreeCache]移除远程节点信息,  node={}", JsonUtil.pojoToJson(remoteNode));
    }
}
